package io.kiki.stack.netty.invoker;


import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.RejectedExecutionException;

/**
 * Listener which listens the Rpc invoke result, and then invokes the call back.
 */

@Slf4j
@Setter
@Getter
public class RpcCallbackListener implements CallbackListener {


    private final String address;

    public RpcCallbackListener(String address) {
        this.address = address;
    }

    @Override
    public void onResponse(Future future) {
        Callback callback = future.getInvokeCallback();
        if (callback != null) {
            CallbackTask callbackTask = new CallbackTask(this.getRemoteAddress(), future);
            if (callback.getExecutor() != null) {
                // There is no need to switch classloader, because executor is provided by user.
                try {
                    callback.getExecutor().execute(callbackTask);
                } catch (RejectedExecutionException e) {
                    if (callback instanceof RejectionProcessableCallback) {
                        switch (((RejectionProcessableCallback) callback).rejectedExecutionPolicy()) {
                        case CALLER_RUNS:
                            callbackTask.run();
                            break;
                        case CALLER_HANDLE_EXCEPTION:
                            callback.onFailed(e);
                            break;
                        case DISCARD:
                        default:
                            log.warn("Callback thread pool busy. discard the callback");
                            break;
                        }
                    } else {
                        log.warn("Callback thread pool busy.");
                    }
                }
            } else {
                callbackTask.run();
            }
        }
    }

    @Override
    public String getRemoteAddress() {
        return this.address;
    }

}
